﻿using EventBus.RabbitMQ.Events;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Collections.Generic;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace EventBus.RabbitMQ
{
    internal sealed class RabbitMqConsumerClient : IConsumerClient
    {
        private readonly IConnectionChannelPool _connectionChannelPool;
        private readonly string _queueName;
        private IModel _channel;
        private IConnection _connection;

        public RabbitMqConsumerClient(string queueName,
            IConnectionChannelPool connectionChannelPool)
        {
            _queueName = queueName;
            _connectionChannelPool = connectionChannelPool;
        }

        public event EventHandler<QueueMessage> OnMessageReceived;

        public void Subscribe(List<RouteInfo> routeInfos)
        {
            if (routeInfos == null)
            {
                throw new ArgumentNullException(nameof(routeInfos));
            }

            Connect(routeInfos);

            foreach (var routeInfo in routeInfos)
            {
                if (!string.IsNullOrWhiteSpace(routeInfo.ExchangeName) && !string.IsNullOrWhiteSpace(routeInfo.RouteKey))
                {
                    _channel.QueueBind(_queueName, routeInfo.ExchangeName, routeInfo.RouteKey);
                }
            }
        }

        public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
        {
            var consumer = new AsyncEventingBasicConsumer(_channel);
            consumer.Received += OnConsumerReceived;

            consumer.Model.BasicQos(0, 100, false);
            consumer.Model.ContinuationTimeout = TimeSpan.FromSeconds(10);
            _channel.BasicConsume(_queueName, false, consumer);

            while (true)
            {
                cancellationToken.ThrowIfCancellationRequested();
                cancellationToken.WaitHandle.WaitOne(timeout);
            }
        }

        public void Commit(object sender)
        {
            _channel.BasicAck((ulong)sender, false);
        }

        public void Reject(object sender)
        {
            _channel.BasicReject((ulong)sender, true);
        }

        public void Dispose()
        {
            _channel?.Dispose();
            _connection?.Dispose();
        }

        public void Connect(List<RouteInfo> routeInfos)
        {
            if (_connection != null) return;
            _connection = _connectionChannelPool.GetConnection();

            _channel = _connection.CreateModel();

            routeInfos.ForEach(x =>
            {
                if (!string.IsNullOrWhiteSpace(x.ExchangeName))
                    _channel.ExchangeDeclare(x.ExchangeName, x.ExchangeType, true);
            });

            _channel.QueueDeclare(_queueName, true, false, false, null);
        }

        private async Task OnConsumerReceived(object sender, BasicDeliverEventArgs e)
        {
            await Task.CompletedTask;
            var headers = new Dictionary<string, string>();
            if (e.BasicProperties.Headers != null)
            {
                foreach (var (key, value) in e.BasicProperties.Headers)
                {
                    headers.Add(key, value == null ? null : Encoding.UTF8.GetString((byte[])value));
                }
            }
            var message = new QueueMessage(_queueName, headers, e.Body.ToArray());

            OnMessageReceived?.Invoke(e.DeliveryTag, message);
        }
    }
}